package cn.getech.data.development.model.vo;

import cn.getech.data.development.constant.FlinkNodeType;
import cn.getech.data.development.constant.YesOrNoEnum;
import cn.getech.data.development.entity.flink.FlinkNodeField;
import cn.getech.data.development.entity.flink.FlinkNodeUnion;
import cn.getech.data.development.entity.flink.FlinkTaskNode;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import org.apache.commons.lang.StringUtils;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Data
public class FlinkTaskNodeVO implements Serializable {
    private static final long serialVersionUID = 1L;

    private Integer id;

    private Integer olderId;

    /**
     * 实时任务id
     */
    private Integer taskId;
    /**
     * 资源id
     */
    private Integer sourceId;
    /**
     * 节点名称
     */
    private String nodeName;

    /**
     * 节点类型 1 source 2 select 3 filter 4 去重 5 group by 6 kafka 7 mysql
     */
    private Integer nodeType;
    private String nodeTypeStr;

    public String getNodeTypeStr() {
        return FlinkNodeType.ObjOf(nodeType).getName();
    }

    /**
     * 英文表名
     */
    private String tableName;

    /**
     * 中文表名
     */
    private String tableAlias;

    /**
     * 目标表名称
     */
    private String targetTableName;

    /**
     * 表达式
     */
    private String expression;

    /**
     * 分区
     */
    private String partitionBy;

    /**
     * 排序
     */
    private String orderBy;

    /**
     * 字段名称
     */
    private String groupBy;

    /**
     * 是否使用窗口函数 1 是,0 否
     */
    private Integer windowsFlag;

    /**
     * 窗口函数字段名称
     */
    private String windowsFieldName;


    /**
     * 间隔时间
     */
    private String interval;

    /**
     * 间隔单位
     */
    private String unit;
    /**
     * "interval": "4",\
     * "unit": "second",\
     * 创建者ID
     */
    private Long createPer;

    /**
     * 创建人
     */
    private String createPerName;

    /**
     * 创建时间
     */
    private Date createTime;

    /**
     * 更新者ID
     */
    private Long updatePer;

    /**
     * 更新时间
     */
    private Date updateTime;
    /**
     * 缓存数据
     */
    private JSONObject config;

    /**
     * 间隔单位
     */
    private String sql;
    /**
     * 字段信息
     */
    private List<FlinkNodeField> fieldList = new ArrayList<>();

    @ApiModelProperty("节点关联信息")
    private List<FlinkNodeUnion> lineList = Lists.newArrayList();

    public FlinkTaskNodeVO() {
    }

    public FlinkTaskNodeVO(FlinkTaskNode node) {
        if (node != null) {
            this.id = node.getId();
            this.sourceId = node.getSourceId();
            this.taskId = node.getTaskId();
            this.nodeName = node.getNodeName();
            this.nodeType = node.getNodeType();
            this.tableName = node.getTableName();
            this.tableAlias = node.getTableAlias();
            if (StringUtils.isNotEmpty(node.getNodeParam())) {
                try {
                    config = JSONObject.parseObject(node.getNodeParam());
                } catch (Exception ignored) {
                }
            }
            this.targetTableName = node.getTargetTableName();
            this.expression = node.getExpression();
            this.partitionBy = node.getPartitionBy();
            this.orderBy = node.getOrderBy();
            this.groupBy = node.getGroupBy();
            this.windowsFlag = node.getWindowsFlag();
            this.windowsFieldName = node.getWindowsFieldName();
            if (StringUtils.isNotEmpty(node.getWindowsInterval())) {
                String[] arr = node.getWindowsInterval().split(",");
                if (arr.length == 2) {
                    this.interval = arr[0];
                    this.unit = arr[1];
                }
            }
            this.createPer = node.getCreatePer();
            this.createTime = node.getCreateTime();
            this.updatePer = node.getUpdatePer();
            this.updateTime = node.getUpdateTime();
        }
    }

    public FlinkTaskNode toDto() {
        String windowsInterval = null;
        if (StringUtils.isNotEmpty(interval) && StringUtils.isNotEmpty(unit)) {
            windowsInterval = interval + "," + unit;
        }
        FlinkTaskNode node = new FlinkTaskNode(taskId, sourceId, nodeName, nodeType, tableName,
                tableAlias, targetTableName, expression, partitionBy, orderBy,
                groupBy, windowsFlag, windowsFieldName, windowsInterval);
        node.setId(this.id);
        if (config != null) {
            node.setNodeParam(config.toJSONString());
        }
        return node;
    }
    public String verificationParam(){
        if (taskId == null) {
            return "任务id不能为空";
        }
        if (nodeType == null) {
            return "节点类型不能为空";
        }
        if (StringUtils.isEmpty(nodeName)) {
            return "节点名称不能为空";
        }
        if (nodeName.length() > 50) {
            return "节点名称长度不能超过50";
        }
        if (StringUtils.isEmpty(tableName)) {
            return "英文表名不能为空";
        }
        if (!tableName.matches("[A-Za-z0-9_]+")) {
            return "英文表名只能包含字母,数字,下划线";
        }
        if (tableName.length() > 200) {
            return "表名长度不能超过200";
        }
//        if (StringUtils.isNotEmpty(tableAlias) && tableAlias.length() > 200) {
//            return "中文表名长度不能超过200";
//        }
//        if (StringUtils.isNotEmpty(targetTableName)) {
//            if (!targetTableName.matches("[A-Za-z0-9_]+")) {
//                return "目标表名只能包含字母,数字,下划线";
//            }
//            if (targetTableName.length() > 200) {
//                return "目标表名长度不能超过200";
//            }
//        }
        if (CollectionUtils.isEmpty(fieldList)) {
            return "字段信息不能为空";
        }
        Set<String> names = fieldList.stream().map(FlinkNodeField::getFieldName).collect(Collectors.toSet());
        if (names.size() != fieldList.size()) {
            return "字段名不能重复";
        }
        Set<String> aliasSet= Sets.newHashSet();
        for (FlinkNodeField flinkNodeField : fieldList) {

            if (nodeType >= 2 && nodeType < 6) {
                if (StringUtils.isNotEmpty(flinkNodeField.getFieldAlias())) {
                    if (!flinkNodeField.getFieldAlias().matches("[A-Za-z0-9_]+")) {
                        return "字段别名只能包含字母,数字,下划线";
                    }
                    if (flinkNodeField.getFieldAlias().length() > 100) {
                        return "字段别名长度不能超过100";
                    }
                }
                if (StringUtils.isNotEmpty(flinkNodeField.getFieldAlias()) &&
                        aliasSet.contains(flinkNodeField.getFieldAlias())) {
                    return "字段别名不能重复";
                }
            }
            aliasSet.add(flinkNodeField.getFieldAlias());
            if (StringUtils.isEmpty(flinkNodeField.getFieldName())) {
                return "字段名不能为空";
            }
            if (!flinkNodeField.getFieldName().matches("[A-Za-z0-9_]+")) {
                return "字段名只能包含字母,数字,下划线";
            }
            if (flinkNodeField.getFieldName().length() > 100) {
                return "字段名长度不能超过100";
            }

            if (StringUtils.isNotEmpty(flinkNodeField.getFieldDesc())) {
                if (flinkNodeField.getFieldDesc().length() > 200) {
                    return "字段描述长度不能超过200";
                }
            }
            if (StringUtils.isNotEmpty(flinkNodeField.getFieldExpression())) {
                if (flinkNodeField.getFieldExpression().length() > 200) {
                    return "字段表达式长度不能超过200";
                }
            }
        }
        if (FlinkNodeType.source.getId().equals(nodeType) ||
                FlinkNodeType.mysql.getId().equals(nodeType) ||
                FlinkNodeType.kafka.getId().equals(nodeType)) {
            if (config == null) {//添加水印
                return "字段配置不能为空";
            }
        }else  if (FlinkNodeType.filter.getId().equals(nodeType)) {
            if (StringUtils.isEmpty(expression)) {
                return "表达式不能为空";
            }
            if (expression.length() > 200) {
                return "表达式长度不能超过200";
            }
        }else  if (FlinkNodeType.group_by.getId().equals(nodeType)) {
            if (StringUtils.isEmpty(groupBy)) {
                return "字段名称不能为空";
            }
            if (YesOrNoEnum.YES.getValue().equals(windowsFlag)) {
                if (StringUtils.isEmpty(windowsFieldName)) {
                    return "窗口字段名称不能为空";
                }
                if (StringUtils.isEmpty(interval)) {
                    return "interval 不能为空";
                }
                if (!interval.matches("[0-9]+")) {
                    return "interval只能为数字";
                }
                if (interval.length() > 10) {
                    return "interval长度不能超过10";
                }
                if (StringUtils.isEmpty(unit)) {
                    return "间隔单位不能为空";
                }
            }
        } else if (FlinkNodeType.unique.getId().equals(nodeType)) {
            if (StringUtils.isEmpty(partitionBy)) {
                return "PARTITION BY不能为空";
            }
            if (partitionBy.length() > 100) {
                return "PARTITION BY 长度不能超过100";
            }
            if (StringUtils.isEmpty(orderBy)) {
                return "ORDER BY不能为空";
            }
        }
        return null;
    }
    public String wrapSQL() {
        StringBuilder sql = new StringBuilder();
       if (FlinkNodeType.select.getId().equals(nodeType)) {
            // SELECT  order_id AS id , FROM source_order；
            sql = new StringBuilder(parseSelect());
            if (StringUtils.isNotEmpty(targetTableName)) {
                sql.append("\n ");
                sql.append(parseInsert(sql.toString()));
            }

        } else if (FlinkNodeType.filter.getId().equals(nodeType)) {
            // SELECT  order_id AS id , FROM source_order  where id=100098 and units=3；
            sql = new StringBuilder(parseSelect());
            if (StringUtils.isNotEmpty(expression)) {
                sql.append(" where ").append(expression);
            }
            if (StringUtils.isNotEmpty(targetTableName)) {
                sql.append("\n ");
                sql.append(parseInsert(sql.toString()));
            }
        } else if (FlinkNodeType.group_by.getId().equals(nodeType)) {
            // SELECT id, GROUP BY  TUMBLE(ts,INTERVAL '4' SECOND), id
            sql = new StringBuilder(parseSelect());
            if (StringUtils.isNotEmpty(groupBy)) {
                if (YesOrNoEnum.YES.getValue().equals(windowsFlag)) {
                    if (StringUtils.isNotEmpty(windowsFieldName)) {
                        sql.append(" GROUP BY  ")
                                .append(" TUMBLE(").append(windowsFieldName)
                                .append(",INTERVAL '").append(interval).append("' ")
                                .append(unit).append("),").append(groupBy);
                    }
                } else {
                    sql.append(" GROUP BY  ").append(groupBy);
                }
            }
            if (StringUtils.isNotEmpty(targetTableName)) {
                sql.append("\n ");
                sql.append(parseInsert(sql.toString()));
            }
        } else if (FlinkNodeType.unique.getId().equals(nodeType)) {
            // SELECT *, ROW_NUMBER() OVER(PARTITION BY ts ORDER BY duration desc) AS row_num  FROM user_log
            sql = new StringBuilder("SELECT ");
            if (CollectionUtils.isNotEmpty(fieldList)) {
                for (FlinkNodeField field : fieldList) {
                    sql.append(field.getFieldName()).append(" AS ").append(field.getFieldAlias()).append(",");
                }
            }
            sql.append(" ROW_NUMBER() OVER(PARTITION BY ")
                    .append(partitionBy)
                    .append(" ORDER BY ")
                    .append(orderBy)
                    .append(" ) as  ")
                    .append("row_num");//别名默认为  row_num
            sql.append(" FROM ");
            sql.append(tableName);
        }
        return sql.toString();
    }

    private StringBuilder parseInsert(String selectSql) {
        StringBuilder sql = new StringBuilder("insert into ");
        sql.append(targetTableName)
                .append(" ").append(selectSql);
        return sql;
    }

    private String parseSelect() {
        StringBuilder sql = new StringBuilder(" SELECT ");
        if (CollectionUtils.isNotEmpty(fieldList)) {
            for (FlinkNodeField field : fieldList) {
                String alias = StringUtils.isEmpty(field.getFieldAlias()) ?
                        field.getFieldName() : field.getFieldAlias();
                if (StringUtils.isNotEmpty(field.getFieldExpression())) {
                    sql.append(field.getFieldExpression())
                            .append(" AS ").append(alias).append(",");
                } else {
                    sql.append(field.getFieldName())
                            .append(" AS ")
                            .append(alias).append(",");
                }
            }
            sql = new StringBuilder(sql.substring(0, sql.length() - 1));
        }
        sql.append(" FROM ").append(tableName);
        return sql.toString();
    }
}
